package com.rc.saas.mini.config;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

/**
 * Created by Sven on 2021/02/09
 */
//@Component
public class EvchargerKafkaConsumer {
    private static Logger logger = LogManager.getLogger(EvchargerKafkaConsumer.class);

    //自定义topic
    public static final String TOPIC_TEST = "command1.test";

    //@Autowired
    //private ConsumerFactory consumerFactory;

    /**
     * 消息过滤器可以在消息抵达consumer之前被拦截，在实际应用中，我们可以根据自己的业务逻辑，
     * 筛选出需要的信息再交由KafkaListener处理，不需要的消息则过滤掉。
     * 配置消息过滤只需要为 监听器工厂 配置一个RecordFilterStrategy（消息过滤策略），
     * 返回true的时候消息将会被抛弃，返回false时，消息能正常抵达监听容器
     * @return
     */
    //@Bean
    //public ConcurrentKafkaListenerContainerFactory filterContainerFactory() {
    //    ConcurrentKafkaListenerContainerFactory factory = new ConcurrentKafkaListenerContainerFactory();
    //    factory.setConsumerFactory(consumerFactory);
    //    // 被过滤的消息将被丢弃
    //    factory.setAckDiscarded(true);
    //    // 消息过滤策略
    //    factory.setRecordFilterStrategy(consumerRecord -> {
    //        if (Integer.parseInt(consumerRecord.value().toString()) % 2 == 0) {
    //            return false;
    //        }
    //        //返回true消息则被过滤
    //        return true;
    //    });
    //    return factory;
    //}

    // 消费监听
    @KafkaListener(topics = {TOPIC_TEST})
    //@KafkaListener(id = "consumer0", groupId = "open-group", topics = TOPIC_TEST)
    public void onMessage(ConsumerRecord<?, ?> record) {
        System.out.println("消费中1............");
        //for (ConsumerRecord<?, ?> record : records) {
        //    System.out.println(record.value());
        //}

        // 消费的哪个topic、partition的消息,打印出消息内容
        System.out.println("简单消费：" + record.topic() + "-" + record.partition() + "-" + record.value());
    }
}
